package com.atguigu.chapter05.sink;

import com.alibaba.fastjson.JSON;
import com.atguigu.bean.WaterSensor;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;
import org.elasticsearch.common.xcontent.XContentType;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

/**
 * Author: Pepsi
 * Date: 2023/7/30
 * Desc:
 */
public class Flink04_Es_1 {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port",1000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> result = env.socketTextStream("hadoop101",9999)
                .map(line->{
                    String[] data = line.split(",");
                    return new WaterSensor(data[0],Long.valueOf(data[1]),Integer.valueOf(data[2]));
                })
                .keyBy(WaterSensor::getId)
                .sum("vc");

        List<HttpHost> hosts = Arrays.asList(
                new HttpHost("hadoop101", 9200),
                new HttpHost("hadoop102", 9200),
                new HttpHost("hadoop103", 9200)
        );

        ElasticsearchSink.Builder<WaterSensor> builder = new ElasticsearchSink.Builder<WaterSensor>(
                hosts,
                new ElasticsearchSinkFunction<WaterSensor>() {
                    @Override
                    public void process(WaterSensor element,  // 需要写出的元素
                                        RuntimeContext runtimeContext, // 运行时上下文   不是context上下文对象
                                        RequestIndexer requestIndexer) {  // 把要写出的数据，封装到RequestIndexer里面
                        String msg = JSON.toJSONString(element);

                        IndexRequest ir = Requests
                                .indexRequest("sensor")
                                .type("_doc")  // 定义type的时候, 不能下划线开头. _doc是唯一的特殊情况
                                .id(element.getId())  // 定义每条数据的id. 如果不指定id, 会随机分配一个id. id重复的时候会更新数据
                                .source(msg, XContentType.JSON);

                        requestIndexer.add(ir);  // 把ir存入到indexer, 就会自动的写入到es中
                    }
                }
        );

        // 自动刷新时间
        builder.setBulkFlushInterval(2000);  // 默认不会根据时间自动刷新
        builder.setBulkFlushMaxSizeMb(1024);  // 当批次中的数据大于等于这个值刷新
        builder.setBulkFlushMaxActions(2);   // 每来多少条数据刷新一次
        // 这三个是或的关系，只要有一个满足就会刷新

        result.addSink(builder.build());

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
/*

index  数据库
type    表(6.X开始type只能有一个, 6.x之前可以有多个. 从7.x开始 type被废弃)
document  行
field value 列

 */